#include <errno.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include "debug_assert.h"
#include "debug_log.h"
#include "linux_socket.h"
#include "server_config.h"
#include "linux_msg.h"
#include "mysql_package.h"
#include "violate_data.h"

#define INPUT_VIOFLAG    100
#define OUTPUT_VIOFLAG   101
#define SOCKET_UNNORMAL  103
#define SOCKET_NORMAL    104

static int g_close_flag = -1;
static int g_heart_time = -1;
static int g_vio_flag = OUTPUT_VIOFLAG;
static t_VioData g_vio_data = {0};
static pthread_mutex_t  g_vio_lock;
static pthread_mutex_t  g_heart_lock;

static bool TransHeader(const int connfd, t_VioHead *p_vio_head) {	
	int recv_ret;

	ASSERT(0 != connfd && NULL != p_vio_head);
	memset(p_vio_head, 0, sizeof(t_VioHead));
	recv_ret = RecvSocket(connfd, p_vio_head, sizeof(t_VioHead));
	if (-1 == recv_ret) {
		LOG("RecvSocket() error! recv_ret=%d\n", recv_ret);
		return false;			
	}
	else {
		if (0 != strcmp(p_vio_head->m_head, "topsky")) {
			LOG("RecvHeader() error! m_head=%s\n", p_vio_head->m_head);
			return false;
		}
		else {
			return true;
		}
	}
}

static bool TransHeartBeat(const int connfd, t_HeartBeat *p_heartbeat) {
	int recv_ret;

	ASSERT(0 != connfd && NULL != p_heartbeat);
	recv_ret = RecvSocket(connfd, p_heartbeat, sizeof(t_HeartBeat));
	if (-1 == recv_ret) {
		LOG("RecvSocket() error! recv_ret=%d\n", recv_ret);	
		return false;			
	}	
	else {
		return true;
	}
}

static bool TransReserved(const int connfd, const t_VioData *p_vio_data) {
	int send_ret;

	ASSERT(0 != connfd && NULL != p_vio_data);
	send_ret = SendSocket(connfd, (t_VioData *)p_vio_data, sizeof(t_VioData));
	if (-1 == send_ret) {
		LOG("SendSocket() error! send_ret=%d\n", send_ret);	
		return false;
	}
	else
		return true;
}

static bool TransSocketData(const int connfd) {
	t_VioHead vio_head = {0};	
	t_HeartBeat heart_beat = {0};
	t_VioData vio_data = {0};
	bool ok_header;
	bool ok_reserved;

	ok_header = TransHeader(connfd, &vio_head);			
	if (false == ok_header) {
		LOG("TransHeader() error!\n");
		return false;
	}
	else {	
		//LOG("TransHeader() success!\n");	
		switch (vio_head.m_type) {
			case VIOHEAD_TYPE_HEARTBEAT:
				ok_reserved = TransHeartBeat(connfd, &heart_beat);
				if (true == ok_reserved) {
					pthread_mutex_lock(&g_heart_lock);
					g_heart_time = heart_beat.m_time;
					//LOG("g_heart_time=%d", g_heart_time);
					pthread_mutex_unlock(&g_heart_lock);						
				}	
				break;
			case VIOHEAD_TYPE_VIODATA:
				pthread_mutex_lock(&g_vio_lock);
				if (INPUT_VIOFLAG == g_vio_flag) {
					memcpy(&vio_data, &g_vio_data, sizeof(t_VioData));	
					ok_reserved = TransReserved(connfd, &vio_data);
					if (false == ok_reserved) {
						LOG("TransReserved() failure!\n");	
					}
					else {
						g_vio_flag = OUTPUT_VIOFLAG;
						LOG("TransReserved() success!\n");
						PrintVioData(&vio_data);	
						//printf("pic_path=%s\n", vio_data.m_pic_path);
						LOG("[voi_data_3]plate=%s,reason=%s,pic_path=%s\n", vio_data.m_plate, vio_data.m_reason, vio_data.m_pic_path);
					}						
				}	
				else {
					memset(&vio_data, 0, sizeof(t_VioData));
					vio_data.m_type = -1;
					ok_reserved = TransReserved(connfd, &vio_data);
					if (false == ok_reserved) {
						LOG("TransReserved-emptyVioData() failure!\n");	
					}
					else {
						//LOG("TransReserved-emptyVioData() success!\n");
					}					
				}
				pthread_mutex_unlock(&g_vio_lock);				
				break;
			default:
				ok_reserved = false;
				break;
		}
		return ok_reserved;
	}
}

static void* PopQueueFunc(void *pv) {
	MYSQL *p_handle = NULL;	
	t_VioData vio_data;
	char config_path[256];
	char mysql_host[32];
	int mysql_port;
	int queue_id = -1;	
	int pop_ret;
	int msg_num;
	bool ok_open;	
	
	GetConfigPath(config_path, sizeof(config_path));
	GetMysqlHost(config_path, mysql_host, sizeof(mysql_host));
	GetMysqlPort(config_path, &mysql_port);
	LOG("Mysql: host=%s, port=%d\n", mysql_host, mysql_port);	
	Mysql_Init(mysql_host, mysql_port, "uploader", "uploader", &p_handle);	
	LOG("Mysql_Init() handle=%p\n", p_handle);
	while (true) {
		if (-1 == queue_id) {
			ok_open = GetMsgQueue("/TOPSKY/LOCAL", &queue_id);
			if (false == ok_open) {
				LOG("GetMsgQueue() error! queue_id=%d\n", queue_id);
			}	
			else {
				//LOG("GetMsgQueue() success! queue_id=%d\n", queue_id);				
			}
		}
		else {
			msg_num = GetMsgNumInQueue(queue_id);
			//LOG("msg_num=%d\n", msg_num);
			if (40 < msg_num) {  // 控制消息队列中消息数量最大为 40
				pop_ret = PopMsgQueue(queue_id, &vio_data, sizeof(vio_data));	
				printf("too mush! PopMsgQueue()! pop_ret=%d\n", pop_ret);	
				if (sizeof(vio_data) == pop_ret && NULL != p_handle 
					&& 0 != strcmp(vio_data.m_reason, "")) { // 布控入 t_alarm 库
					LOG("[voi_data_1]msg_num=%d,plate=%s,reason=%s,pic_path=%s\n", msg_num, vio_data.m_plate, vio_data.m_reason, vio_data.m_pic_path);
					Mysql_WriteAlarm(p_handle, &vio_data);
				}				
			}
			else if (0 < msg_num) {   
				pthread_mutex_lock(&g_vio_lock);
				if (OUTPUT_VIOFLAG == g_vio_flag) {			
					pop_ret = PopMsgQueue(queue_id, &vio_data, sizeof(vio_data));			
					if (sizeof(vio_data) != pop_ret) {
						LOG("PopMsgQueue() error! pop_ret=%d\n", pop_ret);
						queue_id = 0;				
					}
					else {
						printf("PopMsgQueue() success! pop_ret=%d\n", pop_ret);	
						g_vio_flag = INPUT_VIOFLAG;							
						memcpy(&g_vio_data, &vio_data, sizeof(t_VioData));			
						if (NULL != p_handle && 0 != strcmp(vio_data.m_reason, "")) { // 布控入 t_alarm 库
							LOG("[voi_data_2]msg_num=%d,plate=%s,reason=%s,pic_path=%s\n", msg_num, vio_data.m_plate, vio_data.m_reason, vio_data.m_pic_path);
							Mysql_WriteAlarm(p_handle, &vio_data);
						}
					}				
				}
				pthread_mutex_unlock(&g_vio_lock);			
			}
		}
		usleep(1000 * 1000 * 1);		
	}
	Mysql_Uninit(p_handle);
}

static void* PushQueueFunc(void *pv) {
	t_VioData vio_data;
	bool ok_open;
	int queue_id;	
	int push_ret;
	
	while (true) {
		if (0 == queue_id) {
			ok_open = GetMsgQueue("/TOPSKY/LOCAL", &queue_id);
			if (false == ok_open) {
				LOG("GetMsgQueue() error! guding_51002\n");
			}	
		}
		else {
			InitVioData(&vio_data);
			push_ret = PushMsgQueue(queue_id, &vio_data, sizeof(vio_data));
			if (sizeof(vio_data) != push_ret) {
				LOG("PushMsgQueue() error! push_ret=%d\n", push_ret);
				queue_id = 0;				
			}	
		}
		usleep(1000 * 1000 * 2);		
	}
}

static void* HeartBeatFunc(void *pv) {
	int *p_connfd = (int *)pv;
	int pre_time;
	int next_time;
	
	while (true) {
		pthread_mutex_lock(&g_heart_lock);
		pre_time = g_heart_time;
		pthread_mutex_unlock(&g_heart_lock);			
		
		usleep(1000 * 1000 * 13);	// 心跳 13s 一次

		pthread_mutex_lock(&g_heart_lock);	
		next_time = g_heart_time;	
		pthread_mutex_unlock(&g_heart_lock);	
		
		if (pre_time == next_time || SOCKET_UNNORMAL == g_close_flag) {
			pthread_mutex_lock(&g_heart_lock);	
			g_close_flag = SOCKET_UNNORMAL;
			pthread_mutex_unlock(&g_heart_lock);	
			
			CloseSocket(*p_connfd);
			break;
		}
		else {
			pthread_mutex_lock(&g_heart_lock);	
			g_close_flag = SOCKET_NORMAL;
			pthread_mutex_unlock(&g_heart_lock);			
		}
	}
}


#if 1
int main(void) {
	pthread_t popmsg_id;	
	pthread_t pushmsg_id;	
	pthread_t heartbeat_id;	
	char config_path[256];	
	int listenfd;
	int connfd;
	int port;	
	int trans_count;
	int vio_flag;
	bool ok_init;
	bool ok_accept;	
	bool ok_trans;

	LOG(">>start.");
	usleep(1000 * 1000 * 50);	
	
	pthread_mutex_init(&g_heart_lock, NULL);	
	pthread_mutex_init(&g_vio_lock, NULL);		
	
	//pthread_create(&pushmsg_id, NULL, &PushQueueFunc, NULL);	
	pthread_create(&popmsg_id, NULL, &PopQueueFunc, NULL);
	
	GetConfigPath(config_path, sizeof(config_path));
	LOG("config_path=%s\n", config_path);
	GetSocketPort(config_path, &port);
	ok_init = InitSocketServer(port, &listenfd);
	if (false == ok_init) {
		printf("InitSocketServer() error! port=%d, listenfd=%d\n", port, listenfd);		
		LOG("InitSocketServer() error! port=%d, listenfd=%d\n", port, listenfd);
	}
	else {
		printf("InitSocketServer() success! port=%d, listenfd=%d\n", port, listenfd);	
		LOG("InitSocketServer() success! port=%d, listenfd=%d\n", port, listenfd);	
		while (true) {
			ok_accept = AcceptSocket(listenfd, &connfd);
			if (false == ok_accept) {
				printf("AcceptSocket() error! listenfd=%d, connfd=%d\n", listenfd, connfd);
				LOG("AcceptSocket() error! listenfd=%d, connfd=%d\n", listenfd, connfd);
				usleep(1000 * 1000 * 3);
			}
			else {
				printf("AcceptSocket() success! listenfd=%d, connfd=%d\n", listenfd, connfd);					
				LOG("AcceptSocket() success! listenfd=%d, connfd=%d\n", listenfd, connfd);	

				pthread_mutex_lock(&g_vio_lock);
				g_vio_flag = OUTPUT_VIOFLAG;	 // 开始从消息队列中取数据					
				pthread_mutex_unlock(&g_vio_lock);		
				
				pthread_mutex_lock(&g_heart_lock);	
				if (SOCKET_UNNORMAL == g_close_flag || -1 == g_close_flag) {
					pthread_create(&heartbeat_id, NULL, &HeartBeatFunc, &connfd);	
					g_close_flag = SOCKET_NORMAL;
				}
				pthread_mutex_unlock(&g_heart_lock);					
											
				trans_count = 0;
				while (true) {
					//LOG("trans_count:%d\n", trans_count);					
					ok_trans = TransSocketData(connfd);		
					if (false == ok_trans) {
						break;								
					}
					trans_count++;						
					usleep(1000 * 500);					
				}
				
				pthread_mutex_lock(&g_heart_lock);	
				g_close_flag = SOCKET_UNNORMAL;
				pthread_mutex_unlock(&g_heart_lock);

				usleep(1000 * 1000 * 15);  // 确保心跳线程退出				
			}
		}
		CloseSocket(listenfd);
	}	
	
	pthread_mutex_destroy(&g_vio_lock);
	pthread_mutex_destroy(&g_heart_lock);
	
	getchar();
	return 0;
}
#endif
